
/*
 * En este archivo implementamos la funcionalidad de los nodos WORKERS. El comportamiento de los mismos es:
 *
 * Los pasos a llevar cabo por los WORKERS son los siguientes.
 *  --> Esperar instrucciones del MASTER.
 *  --> Si la orden del MASTER es ordenar una nueva lista
 *      <> Obtenemos la lista del MASTER
 *      <> Ordenamos la lista mediante MergeSort en forma local.
 *      <> Solicitar nuevas operaciones al MASTER
 *  --> Si la orden del MASTER es enviar la lista a otro nodo.
 *      <> Obtenemos el nodo destino de la lista.
 *      <> Enviamos la lista ordenada previamente al nodo indicado por el MASTER.
 *      <> Finalizar el bucle.
 *  --> Si la orden del MASTER es recibir una lista de otro nodo.
 *      <> Recibimos la lista del nodo indicado por el MASTER.
 *      <> Ordenamos la lista mdiante MergeSort en forma local.
 *      <> Solicitar nuevas operaciones al MASTER.
 *  --> Si el MASTER indica que no debemos hacer mas nada.
 *      <> Finalizar el bucle
 *  --> Fin de programa
 */

#include "worker.h"
#include "mensajes.h"
#include "../common/ordenamiento.h"
#include <mpi.h>
#include <malloc.h>
#include <string.h>

#define FULL_MERGESORT          0
#define SEMI_MERGESORT          1
#define ESPERANDO_ORDEN         2
#define RECIBIENDO_ORDEN        3
#define RECIBIENDO_NUEVA_LISTA  4
#define RECIBIENDO_LISTA_MERGE  5
#define ENVIANDO_LISTA          6
#define FINALIZANDO             7
#define PROCESANDO              8
#define ENVIANDO_SOLICITUD      9
#define ESPERANDO_DESTINO       10
#define OBTENIENDO_NODO_ORIGEN  11
#define ESPERANDO_LISTA_ORIGEN  12
#define OBTENIENDO_TAMANIO      13

int timestamps1[10000];
int accion1[10000];
int parametros1[10000];
int contadorAcciones1 = 0;
int inicial1;

void registrarEventoWorker(int numeroEvento, int parametro) {
    timestamps1[contadorAcciones1] = MPI_Wtime();
    accion1[contadorAcciones1] = numeroEvento;
    parametros1[contadorAcciones1] = parametro;
    contadorAcciones1++;
}

void escribirDump1() {
    char *nombre = malloc(sizeof(char) * 20);
    int rank;
    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
    sprintf(nombre, "debug-%d.csv", rank);
    FILE *filePointer = fopen(nombre, "w");
    int i;
    for(i=0;i<contadorAcciones1;i++) {
        char *accionName1;
        switch (accion1[i]) {
            case 0: accionName1 = "FULL_MERGESORT";
                break;
            case 1: accionName1 = "SEMI_MERGESORT";
                break;
            case 2: accionName1 = "ESPERANDO_ORDEN";
                break;
            case 3: accionName1 = "RECIBIENDO_ORDEN";
                break;
            case 4: accionName1 = "RECIBIENDO_NUEVA_LISTA";
                break;
            case 5: accionName1 = "RECIBIENDO_LISTA_MERGE";
                break;
            case 6: accionName1 = "ENVIANDO_LISTA";
                break;
            case 7: accionName1 = "FINALIZANDO";
                break;
            case 8: accionName1 = "PROCESANDO";
                break;
            case 9: accionName1 = "ENVIANDO_SOLICITUD";
                break;
            case 10: accionName1 = "ESPERANDO_DESTINO";
                break;
            case 11: accionName1 = "OBTENIENDO_NODO_ORIGEN";
                break;
            case 12: accionName1 = "ESPERANDO_LISTA_ORIGEN";
                break;
            case 13: accionName1 = "OBTENIENDO_TAMANIO";
                break;
        }
        int tiempo;
        if (i==0) {
            tiempo = timestamps1[i]-inicial1;
        } else {
            tiempo = timestamps1[i] - timestamps1[i-1];
        }
        fprintf(filePointer, "%d,%d,%s,%d\n", tiempo, timestamps1[i] - inicial1, accionName1, parametros1[i]);
    }
    fclose(filePointer);
}


// Declaración de estructuras locales
struct DATOS_TAREA {
    int idTarea;
    int cantidadDatos;
};

// Declaración de funciones locales
void enviarSolicitudAlMaster();
struct DATOS_TAREA obtenerOrdenesDelMaster();
int *ordenarNuevoVector(int tamanioVector);
void enviarVector(int tamanioVector, int *vector);
int *recibirOrdenarVector(int *vectorActual, int *tamVectorPropio);
int *juntarVectores(int *tamsVects, int **vectores, int cantVect);

/*
 * Implementación de función definida en la cabecera.
 */
void worker() {

    inicial1 = MPI_Wtime();
    registrarEventoWorker(PROCESANDO, -1);

    // El worker funciona en un bucle continuo hasta que la orden del master es finalizar

    // En estas variable guardaremos una referencia al ultimo vector que procesó el worker y su tamaño
    int *vectorActual;
    int tamVectorActual = 0;

    // En estas variables guarderemos los vectores que vamos ordenando
    int **vectOrd = malloc(sizeof(int *) * 1000);
    int tamanios[1000];
    int contVect = 0;

    int continuarBucle = 1; // Usaremos esta variable para finalizar el bucle
    while (continuarBucle) {

        // Obtenemos la siguiente orden del MASTER
        struct DATOS_TAREA datosTarea = obtenerOrdenesDelMaster();

        // Dependiendo de la tarea asignada, ejecutamos la funcion correspondiente
        switch (datosTarea.idTarea) {

            case ORDENAR_NUEVA_LISTA : {
                // Nos envían un nuevo vector. Lo ordenamos y guardamos, junto con su tamaño
                tamanios[contVect] = datosTarea.cantidadDatos;
                tamVectorActual += datosTarea.cantidadDatos;
                vectOrd[contVect] = ordenarNuevoVector(datosTarea.cantidadDatos);
                contVect++;
                break;
            }

            case ORDENAR_Y_JUNTAR : {

                // Nos envían un nuevo vector. Lo ordenamos y guardamos, junto con su tamaño
                tamanios[contVect] = datosTarea.cantidadDatos;
                tamVectorActual += datosTarea.cantidadDatos;
                vectOrd[contVect] = ordenarNuevoVector(datosTarea.cantidadDatos);
                vectorActual = juntarVectores(tamanios, vectOrd, contVect);
       
                free(vectOrd);
                free(tamanios);
                break;
            }

            case ENVIAR_LISTA : {
                // Nos piden que enviemos nuestro vector a otro nodo.
                // Lo enviamos y liberamos la memoria que ocupaba, y salimos del bucle
                enviarVector(tamVectorActual, vectorActual);
//                free(vectorActual);
                continuarBucle = 0;
                break;
            }

            case RECIBIR_LISTA : {
                // Nos piden que juntemos nuestro vector con otro vector que nos envían.
                int *nuevoVector = recibirOrdenarVector(vectorActual, &tamVectorActual);
                free(vectorActual);
                vectorActual = nuevoVector;
                enviarSolicitudAlMaster();
                break;
            }

            case FINALIZAR : {
                continuarBucle = 0;
                break;
            }

        }
    }
    registrarEventoWorker(FINALIZANDO, -1);
    escribirDump1();
}


/*
 * Envía un mensaje al MASTER solicitándole que se le envíe una nueva tarea
 */
void enviarSolicitudAlMaster() {
    int dummyData = 0;
    registrarEventoWorker(ENVIANDO_SOLICITUD, -1);
    MPI_Send(&dummyData, 1, MPI_INT, 0, ESPERANDO_ORDENES, MPI_COMM_WORLD);
    registrarEventoWorker(PROCESANDO,-1);
}


/*
 * Le informa al MASTER que ya finalizamos nuestro trabajo y que estamos a la espera
 * de una nueva tarea. Se queda esperando la prespuesta del master.
 */
struct DATOS_TAREA obtenerOrdenesDelMaster() {

    // Nos quedamos esperando la respuesta del master.
    MPI_Status mensaje;
    registrarEventoWorker(ESPERANDO_ORDEN, -1);
    MPI_Probe(0, MPI_ANY_TAG, MPI_COMM_WORLD, &mensaje);
    registrarEventoWorker(RECIBIENDO_ORDEN, -1);
    struct DATOS_TAREA datosTarea;
    MPI_Get_count(&mensaje, MPI_INT, &datosTarea.cantidadDatos);
    registrarEventoWorker(PROCESANDO, -1);
    datosTarea.idTarea = mensaje.MPI_TAG;

    // Retornamos la tarea asignada por MASTER, (que se envió en el tag del mensaje)
    return datosTarea;

}


/*
 * Obtiene los datos del mensaje que ha enviado el MASTER previamente.
 * Ordena el vector recibido mediante MergeSort.
 */
int *ordenarNuevoVector(int tamanioVector) {

    // Creamos el buffer para recibir el vector enviado por el MASTER
    int *vectorDesord = malloc(sizeof(int) * tamanioVector);

    // Obtenemos el vector a ordenar del mensaje enviado por el MASTER
    registrarEventoWorker(RECIBIENDO_NUEVA_LISTA,-1);
    MPI_Recv(vectorDesord, tamanioVector, MPI_INT, 0, ORDENAR_NUEVA_LISTA, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
    registrarEventoWorker(FULL_MERGESORT,-1);
    // Ahora tenemos que ordenar el vector recibido.
    int *vectorOrdenado;
    if (tamanioVector > 1) {
        vectorOrdenado = mergesortRecursivo(vectorDesord, tamanioVector);
        free(vectorDesord);
    } else {
        vectorOrdenado = vectorDesord;
    }
    
    return vectorOrdenado;
    registrarEventoWorker(PROCESANDO,-1);

}


/*
 * Envía el vector que se ordenó en la tarea anterior, al nodo indicado
 * en el mensaje que envió el MASTER previamente.
 */
void enviarVector(int tamanioVector, int *vector) {

    // Obtenemos el destino del vector, del mensaje enviado por el MASTER
    int nodoDestino;
    registrarEventoWorker(ESPERANDO_DESTINO,-1);
    MPI_Recv(&nodoDestino, 1, MPI_INT, 0,
            ENVIAR_LISTA, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
    registrarEventoWorker(ENVIANDO_LISTA, nodoDestino);

    int particiones = 1;
    if (tamanioVector > TAM_MAX_PAQUETE) {
        particiones = ceil((tamanioVector * 1.0) / TAM_MAX_PAQUETE;
    }

    int cantBase = tamanioVector / TAM_MAX_PAQUETE;
    int sobrantes = tamanioVector % TAM_MAX_PAQUETE;

    int *tams = malloc(sizeof(int) * particiones);
    int r;
    for(r=0; r<particiones; r++) {
        tams[r] = cantBase;
    }
    for(r=0; r<sobrantes; r++) {
        tams[r] = tams[r] + 1;
    }

    int *offset = vector;
    int p;
    for (p=0; p<particiones; p++) {

        int mensaje;
        if (p==particiones-1) {
            mensaje = TRANSFERIR_LISTA_Y_ORDENAR;
        } else {
            mensaje = TRANSFERIR_LISTA;
        }

        MPI_Request request;
        MPI_ISend(offset, tams[r], MPI_INT, nodoDestino, mensaje, MPI_COMM_WORLD, &request);
        offset += tams[r];
    }

    free(tams);

    registrarEventoWorker(PROCESANDO,-1);

}


/*
 * Obtiene el vector del nodo especificado por el master.
 * Ordena el vetor recibido.
 */
int *recibirOrdenarVector(int *vectorPropio, int *tamVectorPropio) {

    // Obtenemos el nodo del que debemos recibir el vector
    int nodoOrigen;
    MPI_Recv(&nodoOrigen, 1, MPI_INT, 0, RECIBIR_LISTA, MPI_COMM_WORLD, MPI_STATUS_IGNORE);

    int contadorVectores = 0;
    int tams[1000];
    int totalTam = 0;
    int **vectores = malloc(sizeof(int *) * 1000);

     // Aquí guardaremos los vectores recibidos de otros workers
    int puedeJuntar = 0;
    int tamVectorJuntar;
    int *vectorJuntar;

    int continuar = 1;
    while(continuar) {

        // Ahora debemos averiguar la cantidad de datos recibir
        MPI_Status mensaje;
        MPI_Probe(nodoOrigen, MPI_ANY_TAG, MPI_COMM_WORLD, &mensaje);

        int tamVectorRemoto;
        MPI_Get_count(&mensaje, MPI_INT, &tamVectorRemoto);
        totalTam += tamVectorRemoto;

        // Creamos el buffer para recibir el vector enviado por el nodo destino, y lo recibimos
        int *vectorActual = malloc(sizeof(int) * tamVectorRemoto);
        MPI_Recv(vectorActual, tamVectorRemoto, MPI_INT, nodoOrigen, mensaje->MPI_TAG, MPI_COMM_WORLD, MPI_STATUS_IGNORE);

        if (puedeJuntar) {
            vectores[contadorVectores] = juntarVectoresOrdenados(tamVectorRemoto, vectorActual, tamVectorJuntar, vectorJuntar);
            tams[contadorVectores] = tamVectorJuntar + tamVectorRemoto;
            puedeJuntar = 0;
            contadorVectores++;
            free(vectorJuntar);
            free(vectorActual);
        } else {
            if (mensaje->MPI_TAG == TRANSFERIR_LISTA) {
                tamVectorJuntar = tamVectorRemoto;
                vectorJuntar = vectorActual;
                puedeJuntar = 1;
            } else {
                vectores[contadorVectores] = vectorActual;
                tams[contadorVectores] = tamVectorRemoto;
                contadorVectores++;
            }
        }

        if (mensaje->MPI_TAG == TRANSFERIR_LISTA_Y_ORDENAR) {
            continuar = 0;
        }

    }

    int *resultado = juntarVectores(tams, vectores, contadorVectores);
    free(tams);
    free(vectores);

    int resultadoTotal = juntarVectoresOrdenados(totalTam, resultado, tamVectorPropio, vectorPropio);
    free(vectorPropio);
    free(resultado);
    
    return resultadoTotal;
}


int *juntarVectores(int *tamsVects, int **vectores, int cantVect) {

    int vectoresRestantes = cantVect;

    if (vectoresRestantes % 2 == 1) {
        
        int *vector1 = vectores[cantVect-2];
        int *vector2 = vectores[cantVect-1];
        vectores[cantVect-2] = juntarVectoresOrdenados(tamsVects[cantVect-2], vectores[cantVect-2], tamsVects[cantVect-1], vectores[cantVect-1]);
        tamsVects[cantVect-2] = tamsVects[cantVect-2] + tamsVects[cantVect-1];
        free(vector1);
        free(vector2);

        vectoresRestantes--;
    }

    while(vectoresRestantes > 1) {
        int u;
        for (u=0; u<vectoresRestantes; u+=2) {
            int *vector1 = vectores[u];
            int *vector2 = vectores[u+1];
            vectores[u] = juntarVectoresOrdenados(tamsVects[u], vectores[u], tamsVects[u+1], vectores[u+1]);
            tamsVects[u] = tamsVects[u] + tamsVects[u+1];
            free(vector1);
            free(vector2);
        }
        vectoresRestantes /= 2;
    }

    return vectores[0];
}
